package service

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"go-canal/utils/logger"
)

func CanalCustomerMessage(consumerMsg *sarama.ConsumerMessage) error {
	//:读取信息,写入到mysql中
	var (
		msg CanalBody
		err error
	)

	defer func() {
		// 对程序进行recover处理
		if errCover := recover(); errCover != nil {
			logger.Get().Error(fmt.Sprintf("panic kafka canal analysis data err,err is %v,msg = %+v", errCover, string(consumerMsg.Value)))
			Robot.SendMessage("text", fmt.Sprintf("panic kafka canal analysis data err,err is %v,offset = %d,msg = %s", errCover, consumerMsg.Offset, string(consumerMsg.Value)))
		}

		// 如果数据接收过程中有错误,那么记录日志,同时发送消息
		if err != nil {
			logger.Get().Error(fmt.Sprintf("kafka  canal analysis err,err is %v,msg = %+v", err, string(consumerMsg.Value)))
			Robot.SendMessage("text", fmt.Sprintf("kafka canal analysis data err,err is %v,offset = %d,msg = %s", err, consumerMsg.Offset, string(consumerMsg.Value)))
		}
	}()
	fmt.Printf("msg %+v\n", string(consumerMsg.Value))
	err = json.Unmarshal(consumerMsg.Value, &msg)
	if err != nil {
		return err
	}
	fmt.Printf("CanalCustomerMessage consumerMsg,topic = %v,partition = %v,offset = %v,msg=%v,", consumerMsg.Topic, consumerMsg.Partition, consumerMsg.Offset, msg)
	//logger.Get().Info(fmt.Sprintf("CanalCustomerMessage consumerMsg,topic = %v,partition = %v,offset = %v,msg=%v,", consumerMsg.Topic, consumerMsg.Partition, consumerMsg.Offset, msg))
	ctx := context.Background()
	err = parseAnalysisMessage(ctx, msg)
	return err
}
